package handler

import (
	"context"
	"encoding/json"
	"errors"
	"github.com/apache/rocketmq-clients/golang/v5"
	pb "goods-srv/gen/goods"
	"goods-srv/internal/dao"
	"goods-srv/internal/model"
	"goods-srv/libs"
	"google.golang.org/protobuf/types/known/emptypb"
	"log"
	"time"
)

type goodsServer struct {
	pb.UnsafeGoodsServer

	goodsDao dao.IGoods

	simpleConsumer golang.SimpleConsumer
}

// DeductionStock 扣减库存
func (s goodsServer) DeductionStock(_ context.Context, goodsSkuListMessage *pb.GoodsSkuListMessage) (*emptypb.Empty, error) {
	var goodsSkuList []model.GoodsSku
	for _, v := range goodsSkuListMessage.List {
		var goodsSku model.GoodsSku
		libs.MappingStruct(v, &goodsSku)
		goodsSkuList = append(goodsSkuList, goodsSku)
	}
	// 执行扣减库存
	err := s.goodsDao.DeductionStock(goodsSkuList)
	if err != nil {
		return new(emptypb.Empty), err
	}
	return new(emptypb.Empty), nil
}

// IncreaseStock 增加库存
func (s goodsServer) IncreaseStock(_ context.Context, goodsSkuListMessage *pb.GoodsSkuListMessage) (*emptypb.Empty, error) {
	var goodsSkuList []model.GoodsSku
	for _, v := range goodsSkuListMessage.List {
		var goodsSku model.GoodsSku
		libs.MappingStruct(v, &goodsSku)
		goodsSkuList = append(goodsSkuList, goodsSku)
	}
	// 执行增加库存
	err := s.goodsDao.IncreaseStock(goodsSkuList)
	if err != nil {
		return new(emptypb.Empty), err
	}
	return new(emptypb.Empty), nil
}

func NewGoodsServer() pb.GoodsServer {
	// 创建事务消费者
	simpleConsumer, err := libs.NewSimpleConsumer(libs.TransactionTopic, libs.TransactionGroup)
	if err != nil {
		panic(errors.New("事务消息消费者创建失败"))
	}

	// start simpleConsumer
	err = simpleConsumer.Start()
	if err != nil {
		log.Fatal("事务消息消费者启动失败")
	}

	libs.Signal(func(_ error) {
		err = simpleConsumer.GracefulStop()
		if err != nil {
			log.Fatal("优雅退出事务消息监听失败")
			return
		}
	})

	// 创建服务
	s := goodsServer{
		goodsDao:       dao.NewGoods(),
		simpleConsumer: simpleConsumer,
	}

	// 库存归还逻辑
	go func() {
		// maxMessageNum 最大处理消息数量
		var maxMessageNum int32 = 16
		// invisibleDuration should > 20s 不可见持续时间
		var invisibleDuration = time.Second * 20
		for {
			mvs, _ := simpleConsumer.Receive(context.TODO(), maxMessageNum, invisibleDuration)
			//if receiveErr != nil {
			//	log.Printf("没有新消息：%s", receiveErr.Error())
			//	continue
			//}
			if len(mvs) == 0 {
				log.Println("无消息")
				continue
			}
			// ack message
			log.Println("监听到消息，执行库存归还逻辑")
			for _, mv := range mvs {
				_ = simpleConsumer.Ack(context.TODO(), mv)
				//log.Println(string(mv.GetBody()))
				var goodsSkuListMessage pb.GoodsSkuListMessage
				if err = json.Unmarshal(mv.GetBody(), &goodsSkuListMessage); err != nil {
					continue
				}
				// 增加库存
				_, err = s.IncreaseStock(context.TODO(), &goodsSkuListMessage)
				if err != nil {
					log.Printf("%s 库存归还失败：%s", goodsSkuListMessage.OrderId, err.Error())
					continue
				}
				log.Printf("%s 库存归还成功", goodsSkuListMessage.OrderId)
			}
		}
	}()

	return s
}
